package com.shenhaofeng.rxbus;


import android.os.Looper;

import rx.Observable;
import rx.Scheduler;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.functions.Func1;
import rx.subjects.PublishSubject;
import rx.subjects.SerializedSubject;
import rx.subjects.Subject;


/**
 * 一个线程不安全的事件总线类，该类的方法不保证线程安全，多线程操作可能导致数据丢失,尽量执行在同一线程上
 * Created by shenhaofeng on 2016/7/18.
 */
public class RxBus {


    //全局实例
    private static volatile RxBus mInstance;

    //工作线程的Looper
    private Looper mLooper;

    //这是个Subject的实例，该类的功能都是围绕它实现的
    private final Subject<Object, Object> subject = new SerializedSubject<>(PublishSubject.create());


    /**
     * 初始化时传入Looper,禁止用户自行实例化
     * <br>必须通过{@link #getInstance()}或者{@link #createInstance(Looper)}来获取或创建实例
     *
     * @param looper 工作线程的Looper
     */
    private RxBus(Looper looper) {
        //禁止外部实例化
        this.mLooper = looper;
    }


    /**
     * 获取全局的bus默认实例,该实例工作在主线程,请不要在其中执行耗时方法或调用网络相关的操作
     *
     * @return 全局主线程的实例
     */
    public synchronized static RxBus getInstance() {
        if (mInstance == null) {
            //默认运行在主线程
            mInstance = new RxBus(Looper.getMainLooper());
        }
        return mInstance;
    }


    /**
     * 通过这个方法可以创建一个新的RxBus对象,需要自行维护对象的引用,不同的RxBus之间无法通讯
     *
     * @return
     */
    public static RxBus createInstance(Looper looper) {
        return new RxBus(looper);
    }


    /**
     * 获取数据源,需要注意的是，尽量保证observe和subscribe在统一线程运行，否则在线程尚未启动时可能会丢失事件,subscribeOn和observeOn已配置，请勿再次配置！
     * <p> 该方法有两个参数来过滤事件，flag用于区分各个事件而Class对象用于匹配事件中符合类型的数据，否则即使接收到了flag，但是数据类型不同可能会导致不可预知的错误</p>
     *
     * @param <D> 需要观察的数据类型
     * @return 一个可观察的数据源对象
     */
    public <D> Observable<D> observable(final String flag, final Class<D> cls) {
        return subject
                //过滤不是event的数据
                .cast(Event.class)
                //过滤不符合标识的数据
                .filter(new Func1<Event, Boolean>() {
                    @Override
                    public Boolean call(Event event) {
                        if (event.getFlag() == null || flag == null || event.getFlag().isEmpty() || flag.isEmpty()) {
                            return false;
                        }
                        return event.getFlag().equals(flag);
                    }
                })
                //转化data,如果数据为null将会直接返回null,如果数据类型错误将会抛出异常
                .concatMap(new Func1<Event, Observable<? extends D>>() {
                    @Override
                    public Observable<? extends D> call(Event event) {
                        Object data = event.getData();
                        if (data == null) {
                            return Observable.just(null);
                        }
                        if (cls.isInstance(data)) {
                            return Observable.just(cls.cast(data));
                        } else {
                            return Observable.error(new EventException("标记和数据类型不匹配，请确认发送的数据和接受的数据what和data相匹配"));
                        }
                    }
                }).subscribeOn(subscribeOn())
                .observeOn(subscribeOn());
    }


    /**
     * 订阅事件
     *
     * @param cls    提供事件数据类型
     * @param action 事件处理
     * @param <T>    事件类型
     * @return 一个Subscription对象，可供取消订阅，自行调用{@link Subscription#unsubscribe()}
     * @see RxBus#observable(String, Class)
     */
    public <T> Subscription subscribe(String flag, Class<T> cls, Action1<T> action) {
        return observable(flag, cls).subscribeOn(subscribeOn()).observeOn(subscribeOn()).subscribe(action);
    }


    /**
     * 订阅事件
     *
     * @param cls        提供事件数据类型
     * @param subscriber 事件处理
     * @param <T>        事件类型
     * @return 一个Subscription对象，可供取消订阅，自行调用{@link Subscription#unsubscribe()}
     * @see RxBus#observable(String, Class)
     */
    public <T> Subscription subscribe(String flag, Class<T> cls, Subscriber<T> subscriber) {
        checkThread();
        return observable(flag, cls).subscribeOn(subscribeOn()).observeOn(subscribeOn()).subscribe(subscriber);
    }


    /**
     * 触发事件,发送一个数据,这个数据将会根据其类型发送到订阅的事件订阅者手中
     *
     * @param event
     */
    public void post(Event event) {
        checkThread();
        subject.onNext(event);
    }


    /**
     * 检查并确保线程安全
     */
    private void checkThread() {
        if (mLooper != Looper.myLooper()) {
            //非同线程，直接抛出异常
            throw new UnsupportedOperationException("RxBus的操作必须在初始化是的指定Looper的线程下，如果是使用的默认getInstance()方法获取的实例，那么所有方法必须运行在主线程。");
        }
    }


    private Scheduler subscribeOn() {
        return AndroidSchedulers.from(mLooper);
    }


}
